package com.howso.cloud.kafka;

import com.howso.cloud.kafka.factory.KafkaConsumerFactory;
import com.howso.cloud.kafka.factory.KafkaFactory;
import com.howso.cloud.kafka.factory.KafkaProducerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;


@Configuration
public class KafkaConfiguration {


    @Bean
    @ConditionalOnMissingBean(KafkaFactory.class)
    public KafkaFactory kafkaFactory(KafkaProperties properties, KafkaTemplate<?, ?> kafkaTemplate,
                                     ConcurrentKafkaListenerContainerFactory<?, ?> containerFactory,
                                     ApplicationContext applicationContext,
                                     ProducerFactory<?, ?> producerFactory,
                                     ConsumerFactory<?, ?> consumerFactory) {
        KafkaFactory kafkaFactory = new KafkaFactory();
        kafkaFactory.setApplicationContext(applicationContext);
        kafkaFactory.setKafkaProperties(properties);
        kafkaFactory.setKafkaTemplate(kafkaTemplate);
        kafkaFactory.setContainerFactory(containerFactory);
        kafkaFactory.setConsumerFactory((KafkaConsumerFactory<?, ?>) consumerFactory);
        kafkaFactory.setProducerFactory((KafkaProducerFactory<?, ?>) producerFactory);
        kafkaFactory.init();
        return kafkaFactory;
    }

    @Bean
    @ConditionalOnMissingBean(ProducerFactory.class)
    public ProducerFactory<?, ?> kafkaProducerFactory(KafkaProperties properties) {
        KafkaProducerFactory<?, ?> factory = new KafkaProducerFactory<>(
                properties.buildProducerProperties());
        String transactionIdPrefix = properties.getProducer().getTransactionIdPrefix();
        if (transactionIdPrefix != null) {
            factory.setTransactionIdPrefix(transactionIdPrefix);
        }
        return factory;
    }

    @Bean
    @ConditionalOnMissingBean(ConsumerFactory.class)
    public ConsumerFactory<?, ?> kafkaConsumerFactory(KafkaProperties properties) {
        return new KafkaConsumerFactory<>(properties.buildConsumerProperties());
    }
}
